package com.borened.netty.protocol.mqtt;

import com.borened.netty.util.ByteBufUtil;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;

/**
 * @author chengcaihua
 * @description 消息发布器
 * @since 2023-11-13 16:56
 */
@Slf4j
@Component
public class MqttPublisher {

    @Resource
    private SubscriberManager subscriberManager;

    /**
     * 保留消息
     */
    private final Map<String, List<MqttPersistentMessage>> retainedMessages = new HashMap<>();

    public Map<String, List<MqttPersistentMessage>> getRetainedMessages() {
        return retainedMessages;
    }

    public void publicQos0(String topic, MqttPublishMessage message) {
        Set<Subscriber> subscribers = subscriberManager.getSubscribers(topic);

        MqttPersistentMessage persistentMessage = MqttPersistentMessage.builder()
                .topic(topic)
                .payload(ByteBufUtil.readBytes(message.payload()))
                .qos(message.fixedHeader().qosLevel())
                .retain(false)
                .timestamp(System.currentTimeMillis())
                .build();
        if (subscribers.size() > 0) {
            for (Subscriber subscriber : subscribers) {
                subscriberManager.executeEvent(()-> subscriber.receiveMessage(persistentMessage));
            }
        }
        //无订阅者，选择保留或丢弃消息
        boolean retain = message.fixedHeader().isRetain();
        if (retain) {
            log.info("retain message, topic:{}, message:{}", topic, message);

            retainedMessages.compute(topic, (key, value) -> {
                if (value == null) {
                    List<MqttPersistentMessage> list = new ArrayList<>();
                    list.add(persistentMessage);
                    return list;
                } else {
                    value.add(persistentMessage);
                    return value;
                }
            });
        }
    }
}
